/**
 * Copyright 2010 Creative Works, Inc.
 * Creative Works licenses this file to you under the Apache License, version
 * 2.0 (the "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *    http://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied.  See the License for the specific language governing
 * permissions and limitations under the License.
 */
package org.sendvox.freswitch.core.impl;

import java.net.ConnectException;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandler.Sharable;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.sendvox.freswitch.event.ChannelAnswerMessage;
import org.sendvox.freswitch.event.ChannelParkMessage;
import org.sendvox.freswitch.event.Event;
import org.sendvox.freswitch.event.impl.EventSocketDispatcher;
import org.sendvox.freswitch.event.impl.EventSocketMessageListener;
import org.sendvox.freswitch.event.impl.EventSocketMessageSource;
import org.sendvox.freswitch.event.impl.Header;
import org.sendvox.freswitch.event.impl.Message;
import org.sendvox.freswitch.event.impl.MessageFactory;
import org.sendvox.freswitch.event.impl.RawMessage;
import org.sendvox.freswitch.inbound.impl.ResponseCallbacks;
import org.sendvox.freswitch.util.Arguments;

@Sharable
public abstract class AbstractChannelHandler extends
		SimpleChannelUpstreamHandler implements
		EventSocketMessageSource<Message> {

	private static final Logger logger = Logger
			.getLogger(AbstractChannelHandler.class.getName());

	static final String MESSAGE_TERMINATOR = "\n\n";
	static final String LINE_TERMINATOR = "\n";

	private final Lock lock = new ReentrantLock();
	private final Queue<Callback> callbacks = new ConcurrentLinkedQueue<Callback>();

	protected final static AtomicLong transferredBytes = new AtomicLong();
	protected final EventSocketDispatcher<Message> dispatcher = new EventSocketDispatcher<Message>();

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {

		if (e.getMessage() instanceof RawMessage) {

			RawMessage response = (RawMessage) e.getMessage();

			Arguments.INSTANCE.validate(response).isNotNull().isNotEmpty();

			logger.log(Level.INFO, "Message received: " + response);

			final String contentType = response.getContentType();
			if (contentType != null) {

				if (contentType.equals(ResponseCallbacks.Value.API_RESPONSE
						.getLiteral())
						|| contentType
								.equals(ResponseCallbacks.Value.COMMAND_REPLY
										.getLiteral())) {

					try {
						getCallbacks().poll().handle(response);
					} catch (Exception re) {
						re.printStackTrace();
					}

				} else if (!response.getHeaders().isEmpty()
						&& contentType
								.equals(ResponseCallbacks.Value.TEXT_EVENT_PLAIN
										.getLiteral())) {

					String eventName = response.getEventBodyMap().get(
							Header.EVENT_NAME);
					Event eventType = Event.fromLiteral(eventName);
					switch (eventType) {
					case CHANNEL_ANSWER:
						ChannelAnswerMessage channelAnswerMsg = MessageFactory.INSTANCE
								.createChannelAnswerMessage(response);
						dispatcher.fire(channelAnswerMsg);
						break;
					case CHANNEL_PARK:
						ChannelParkMessage channelParkMsg = MessageFactory.INSTANCE
								.createChannelParkMessage(response);
						dispatcher.fire(channelParkMsg);
						break;
					default:
						logger.log(
								Level.INFO,
								"Unhandled event: "
										+ response.getEventBodyMap().get(
												Header.EVENT_NAME));
						break;
					}
				}

				handleEventSocketMessage(ctx, response);
			}
		}
	}

	@Override
	public void addEventSocketListener(
			EventSocketMessageListener<Message> messageHandler) {
		dispatcher.addListener(messageHandler);
	}

	@Override
	public void removeEventSocketListener(
			EventSocketMessageListener<Message> messageHandler) {
		dispatcher.removeListener(messageHandler);
	}

	public static long getTransferredBytes() {
		return transferredBytes.get();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent ee) {
		// Close the connection when an exception is raised.
		if (ee.getCause() instanceof ConnectException) {
			logger.log(Level.SEVERE, "Error connecting to FreeSWITCH server",
					ee.getCause());

		} else {
			logger.log(Level.SEVERE, "Unexpected exception from downstream.",
					ee.getCause());
		}

		ee.getChannel().close();
	}

	public RawMessage executeSyncCommand(Channel channel, String inputCommand) {
		return execute(channel, inputCommand, true, true);
	}

	public RawMessage executeSyncMultiLineCommand(Channel channel,
			List<String> command) {
		// Build command with double line terminator at the end
		StringBuilder inputCommand = new StringBuilder();
		for (String commandLine : command) {
			inputCommand.append(commandLine).append(LINE_TERMINATOR);
		}
		inputCommand.append(LINE_TERMINATOR);
		return execute(channel, inputCommand.toString(), true, false);
	}

	private RawMessage execute(Channel channel, String inputCommand,
			boolean sync, boolean setMsgTerminator) {
		Arguments.INSTANCE.validate(channel, inputCommand).isNotNull();
		if (sync) {
			Callback callback = new Callback();
			lock.lock();
			try {
				callbacks.add(callback);
				logger.log(Level.INFO, "Sending command $" + inputCommand);
				String cmd = setMsgTerminator ? inputCommand
						.concat(MESSAGE_TERMINATOR) : inputCommand;
				channel.write(cmd);
			} finally {
				lock.unlock();
			}
			return callback.get();
		} else {
			return null;
		}
	}

	protected abstract void handleEventSocketMessage(ChannelHandlerContext ctx,
			RawMessage response);

	protected static class Callback {

		private final CountDownLatch latch = new CountDownLatch(1);

		private RawMessage response;

		RawMessage get() {
			try {
				latch.await();
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
			return response;
		}

		public void handle(RawMessage response) {
			this.response = response;
			latch.countDown();
		}
	}

	/**
	 * @return the callbacks
	 */
	public Queue<Callback> getCallbacks() {
		return callbacks;
	}

}